package com.fulu.game.bi.redis.service.impl;


import com.fulu.game.bi.redis.enums.RedisKeyEnum;
import com.fulu.game.bi.redis.service.RedisOpenService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.dao.DataAccessException;
import org.springframework.data.redis.connection.RedisConnection;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.connection.RedisStringCommands;
import org.springframework.data.redis.core.*;
import org.springframework.data.redis.serializer.RedisSerializer;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;
import org.springframework.util.StopWatch;

import java.util.*;
import java.util.concurrent.TimeUnit;

/**
 * redis操作封装类
 */
@Service
@Slf4j
public class RedisOpenServiceImpl implements RedisOpenService {

    public static final String LOCK_PREFIX = "redis_lock";
    /**
     * 默认存活时间30分钟
     */
    private static final long TIME = 30 * 60;
    private static final int DEFAULT_SCAN_COUNT = 100000;
    private static final String LOCKED = "LOCKED";
    //    private static Lock lock = new ReentrantLock();//基于底层IO阻塞考虑
    @Autowired
    private RedisTemplate redisTemplate;

    /**
     * 获取某个key的值
     *
     * @param key
     * @return
     */
    @Override
    public String get(String key) {
        Object value = redisTemplate.opsForValue().get(key);
        return (value != null) ? value.toString() : null;
    }

    /**
     * 获取某个key的过期时间
     *
     * @param key
     * @return
     */
    @Override
    public Long getExpireTime(String key) {
        Long expireTime = redisTemplate.getExpire(key);
        return (expireTime != null && expireTime > 0) ? expireTime : 0L;
    }


    /**
     * 统计bit位为1的总数
     *
     * @param key
     */
    @Override
    public Long bitCount(final String key) {
        return setForSize(key);
    }

    /**
     * 添加hyperLogLog
     *
     * @param key
     * @param vals
     * @return true添加成功(不存在), false添加失败(已存在)
     */
    public boolean hyperLogLogAdd(String key, Object... vals) {
        return redisTemplate.opsForHyperLogLog().add(key, vals) > 0;
    }

    public boolean hyperLogLogAdd(String key, Long time, Object... vals) {
        boolean res = redisTemplate.opsForHyperLogLog().add(key, vals) > 0;
        if (res) {
            expire(key, time);
        }
        return res;
    }

    /**
     * 计算hyperLogLog大小
     *
     * @param key
     * @return
     */
    public Long hyperLogLogSize(String key) {
        return redisTemplate.opsForHyperLogLog().size(key);
    }


    /**
     * bit 操作
     *
     * @param op
     * @param destination
     * @param keys
     * @return
     */
    @Override
    public Long bitOp(final RedisStringCommands.BitOperation op, final String destination, final String... keys) {
        return (Long) redisTemplate.execute(new RedisCallback<Long>() {
            @Override
            public Long doInRedis(RedisConnection connection) throws DataAccessException {
                long result = 0;
                byte[][] byteArrs = new byte[keys.length][];
                for (int i = 0; i < keys.length; i++) {
                    byteArrs[i] = keys[i].getBytes();
                }
                result = connection.bitOp(op, destination.getBytes(), byteArrs);
                return result;
            }
        });
    }


    /**
     * 添加一个BitSet
     *
     * @param key
     * @param val
     */
    @Override
    public void bitSet(String key, long val, boolean flag) {
        if (flag) {
            setForAdd(key, val);
        } else {
            setForDel(key, val);
        }
    }

    /**
     * 添加一个BitSet,默认为true
     *
     * @param key
     * @param val
     */
    @Override
    public void bitSet(String key, long val) {
        bitSet(key, val, true);
    }

    /**
     * 判断bitSet值(Value是在bit中存在)
     *
     * @param key
     * @param val
     * @return
     */
    @Override
    public Boolean getBitSet(String key, long val) {
        return isMember(key, val);
    }

    /**
     * 设置某个key的值
     *
     * @param key
     * @param value
     */
    @Override
    public void set(String key, String value) {
        set(key, value, TIME);
    }

    /**
     * 设置某个key的值及保存时间
     *
     * @param key
     * @param value
     * @param time
     */
    @Override
    public void set(String key, String value, long time) {
        set(key, value, time, TimeUnit.SECONDS);
    }

    /**
     * string类型存储并设置过期时长
     *
     * @param key
     * @param value
     * @param time
     * @param unit
     */
    @Override
    public void set(String key, String value, long time, TimeUnit unit) {
        redisTemplate.opsForValue().set(key, value, time, unit);
    }

    /**
     * 设置某个key的值
     *
     * @param key
     * @param value
     * @param isPerpetual 是否永久保存（true：是；false：否）
     */
    @Override
    public void set(String key, String value, boolean isPerpetual) {
        if (isPerpetual) {
            redisTemplate.opsForValue().set(key, value);
        } else {
            set(key, value);
        }
    }

    /**
     * 根据key设置某个hash的值
     *
     * @param key
     * @param hash
     * @param value
     */
    @Override
    public void hset(String key, String hash, Object value) {
        hset(key, hash, value, TIME);
    }

    /**
     * 根据key设置某个hash的值
     *
     * @param key
     * @param hash
     * @param value
     * @param isPerpetual 是否永久保存（true：是；false：否）
     */
    @Override
    public void hset(String key, String hash, Object value, boolean isPerpetual) {
        if (isPerpetual) {
            redisTemplate.opsForHash().put(key, hash, value);
        } else {
            hset(key, hash, value, TIME);
        }
    }

    /**
     * 根据key设置某个hash的值及存活时间
     *
     * @param key
     * @param hash
     * @param value
     * @param time
     */
    @Override
    public void hset(String key, String hash, Object value, long time) {
        redisTemplate.opsForHash().put(key, hash, value);
        redisTemplate.expire(key, time, TimeUnit.SECONDS);
    }

    /**
     * 根据key设置hashtable的值
     *
     * @param key
     * @throws Exception
     */
    @Override
    public void hset(String key, Map<String, Object> map) {
        hset(key, map, TIME);
    }

    /**
     * 根据key设置hashtable的值
     *
     * @param key
     * @param map
     * @param isPerpetual 是否永久保存（true：是；false：否）
     */
    @Override
    public void hset(String key, Map<String, Object> map, boolean isPerpetual) {
        if (isPerpetual) {
            redisTemplate.opsForHash().putAll(key, map);
        } else {
            hset(key, map, TIME);
        }
    }

    /**
     * 根据key设置hashtable的值
     *
     * @param key
     * @param time
     * @throws Exception
     */
    @Override
    public void hset(String key, Map<String, Object> map, long time) {
        redisTemplate.opsForHash().putAll(key, map);
        redisTemplate.expire(key, time, TimeUnit.SECONDS);
    }

    /**
     * 删除hash指定的hash keys
     *
     * @param key
     * @param hashKeys
     * @return
     */
    @Override
    public long hdel(String key, Object... hashKeys) {
        return redisTemplate.opsForHash().delete(key, hashKeys);
    }

    /**
     * 根据key删除某个hash的值
     *
     * @param key
     * @param hash
     * @return
     */
    @Override
    public String hdel(String key, String hash) {
        Object value = redisTemplate.opsForHash().delete(key, hash);
        return (value != null) ? value.toString() : null;
    }

    @Override
    public Double hincr(String key, String hashkey, double incr) {
        return redisTemplate.opsForHash().increment(key, hashkey, incr);
    }

    @Override
    public void expire(String key, long time, TimeUnit timeUnit) {
        redisTemplate.expire(key, time, timeUnit);
    }

    @Override
    public void expire(String key, long time) {
        redisTemplate.expire(key, time, TimeUnit.SECONDS);
    }

    @Override
    public void expireAt(String key, Date date) {
        redisTemplate.expireAt(key, date);
    }

    /**
     * 根据key获取某个hash的值
     *
     * @param key
     * @param hash
     * @return
     */
    @Override
    public String hget(String key, String hash) {
        Object value = redisTemplate.opsForHash().get(key, hash);
        return (value != null) ? value.toString() : null;
    }

    /**
     * 根据key获取整个hashtable
     *
     * @param key
     * @return
     */
    @Override
    public Map<String, Object> hget(String key) {
        return redisTemplate.opsForHash().entries(key);
    }

    /**
     * 根据key删除整个hashtable
     *
     * @param key
     * @return
     */
    @Override
    public void delete(String key) {
        redisTemplate.delete(key);
    }

    /**
     * 删除多个key
     */
    @Override
    public void delete(Collection keys) {
        redisTemplate.delete(keys);
    }


    /**
     * 判断key是否存在
     *
     * @param key
     * @return
     */
    @Override
    public boolean hasKey(String key) {
        return redisTemplate.hasKey(key);
    }

    /**
     * redis锁
     *
     * @param key
     * @param expireTime
     */
    @Override
    public boolean lock(String key, long expireTime) {
        if (redisTemplate.opsForValue().setIfAbsent(key, LOCKED)) {
            redisTemplate.expire(key, expireTime, TimeUnit.SECONDS);
            return true;
        }
        return false;
    }

    /**
     * 释放锁
     *
     * @param key
     */
    @Override
    public void unlock(String key) {
        // 释放锁
        redisTemplate.delete(LOCK_PREFIX + key);
    }

    /**
     * 存储时间间隔
     *
     * @param key
     * @param time
     */
    @Override
    public void setTimeInterval(String key, long time) {
        set(RedisKeyEnum.BI_TIME_INTERVAL_KEY.generateKey(key), new Date().getTime() + "", time);
    }


    /**
     * 判断时间间隔
     *
     * @param key
     * @return
     */
    @Override
    public Boolean isTimeIntervalInside(String key) {
        return hasKey(RedisKeyEnum.BI_TIME_INTERVAL_KEY.generateKey(key));
    }

    /**
     * 获取listOps来操作list
     *
     * @param key
     * @return
     */
    @Override
    public <T, S> BoundListOperations<T, S> getListOps(String key) {
        return redisTemplate.boundListOps(key);
    }


    /**
     * 初始化list
     *
     * @param <T>
     * @return
     */
    @Override
    public <T> long listForInit(String key, List<T> values) {
        return redisTemplate.opsForList().rightPushAll(key, values);
    }


    /**
     * 向list某个下标插入元素
     *
     * @param key
     * @param index
     * @param value
     * @param <T>
     */
    @Override
    public <T> void listSetForIndex(String key, int index, T value) {
        redisTemplate.opsForList().set(key, index, value);
    }

    @Override
    public <T> void listSetForRightPush(String key, T value) {
        redisTemplate.opsForList().rightPush(key, value);
    }

    /**
     * 查询list的大小
     *
     * @param key
     * @param <T>
     * @return
     */
    @Override
    public <T> long listForSize(String key) {
        return redisTemplate.opsForList().size(key);
    }


    /**
     * 查询list所有元素
     *
     * @param key
     * @param <T>
     * @return
     */
    @Override
    public <T> List<T> listForAll(String key) {
        try {
            return redisTemplate.opsForList().range(key, 0, -1);
        } catch (Exception e) {
            log.error("redis listForAll error:", e);
            return Collections.emptyList();
        }
    }

    /**
     * 查询list所有元素
     *
     * @param key
     * @param <T>
     * @return
     */
    @Override
    public <T> List<T> listForAll(String key, RedisSerializer<T> redisSerializer) {
        try {
            redisTemplate.setValueSerializer(redisSerializer);
            return redisTemplate.opsForList().range(key, 0, -1);
        } catch (Exception e) {
            log.error("redis listForAll error:", e);
            return Collections.emptyList();
        }
    }


    /**
     * 查询list所有元素
     *
     * @param key
     * @param <T>
     * @return
     */
    @Override
    public <T> T listGetForIndex(String key, int index) {
        try {
            return (T) redisTemplate.opsForList().index(key, index);
        } catch (Exception e) {
            return null;
        }
    }

    @Override
    public <T> List<T> listForRange(String key, int startIndex, int endIndex) {
        return redisTemplate.opsForList().range(key, startIndex, endIndex);
    }

    /**
     * 往SET中插入元素
     *
     * @param key
     * @param value
     * @param <T>
     */
    @Override
    public <T> long setForAdd(String key, T value) {
        SetOperations setOperations = redisTemplate.opsForSet();
        return setOperations.add(key, value);
    }

    @Override
    public long setForInit(String key, List<String> value) {
        return redisTemplate.opsForSet().add(key, value);
    }

    @Override
    public boolean addSetValueCache(String key, String value) {
        try {
            return redisTemplate.opsForSet().add(key, value) > 0;
        } catch (RuntimeException ex) {
            log.warn("add set value cache fail. key:{}, value:{}", key, value, ex);
        }
        return false;
    }

    /**
     * 在SET中删除元素
     *
     * @param key
     * @param value
     * @param <T>
     */
    @Override
    public <T> long setForDel(String key, T value) {
        SetOperations setOperations = redisTemplate.opsForSet();
        return setOperations.remove(key, value);
    }

    /**
     * 查询SET的size
     *
     * @param key
     */
    @Override
    public long setForSize(String key) {
        SetOperations setOperations = redisTemplate.opsForSet();
        return setOperations.size(key);
    }

    /**
     * 查询SET的所有元素
     *
     * @param key
     */
    @Override
    public <T> Set<T> setForAll(String key) {
        SetOperations setOperations = redisTemplate.opsForSet();
        long length = setOperations.size(key);
        if (length <= 0) {
            return new HashSet();
        }
        return setOperations.members(key);
    }

    /**
     * 随机出栈集合的元素
     *
     * @param key 集合key
     * @return 取出的元素
     */
    @Override
    public String randomPop(String key) {
        SetOperations setOperations = redisTemplate.opsForSet();
        Object member = setOperations.randomMember(key);
        if (member == null) {
            return null;
        }
        setOperations.remove(key, member);
        return member.toString();
    }

    /**
     * 判断元素是否存在
     *
     * @param key    集合的key
     * @param member 集合是否存在的member
     * @return 是否存在的结果
     */
    @Override
    public boolean isMember(Object key, Object member) {
        SetOperations setOperations = redisTemplate.opsForSet();
        Boolean result = setOperations.isMember(key, member);
        if (result == null) {
            return false;
        }
        return result;
    }


    /**
     * 往ZSET中插入元素
     *
     * @param key
     * @param value
     * @param score 排序分数
     * @param <T>
     */
    @Override
    public <T> boolean zsetForAdd(String key, T value, double score) {
        ZSetOperations setOperations = redisTemplate.opsForZSet();
        return setOperations.add(key, value, score);
    }

    /**
     * 往ZSET中插入元素(按照集合元素size来定义顺序)
     *
     * @param key
     * @param value
     * @param <T>
     */
    @Override
    public <T> boolean zsetForAdd(String key, T value) {
        ZSetOperations setOperations = redisTemplate.opsForZSet();
        long score = 1;
        if (hasKey(key)) {
            score = zsetForSize(key) + 1;
        }
        return setOperations.add(key, value, score);
    }


    @Override
    public <T> boolean zsetForAdd(String key, T value, long score) {
        ZSetOperations setOperations = redisTemplate.opsForZSet();
        return setOperations.add(key, value, score);
    }

    /**
     * 在ZSET中删除元素
     *
     * @param key
     * @param value
     * @param <T>
     */
    @Override
    public <T> long zsetForDel(String key, T value) {
        ZSetOperations setOperations = redisTemplate.opsForZSet();
        return setOperations.remove(key, value);
    }

    /**
     * 返回元素在集合的排名
     *
     * @param key
     * @param o
     * @return
     */
    @Override
    public Long zsetRank(Object key, Object o) {
        return redisTemplate.opsForZSet().rank(key, o);
    }

    /**
     * 返回元素在集合的排名
     *
     * @param key
     * @param o
     * @return
     */
    @Override
    public Long zsetReverseRank(Object key, Object o) {
        return redisTemplate.opsForZSet().reverseRank(key, o);
    }

    /**
     * 删除ZSET指定SCORE范围内的元素
     *
     * @param key
     * @param start
     * @param end
     * @return
     */
    @Override
    public Long zsetRemoveRangeByScore(Object key, double start, double end) {
        return redisTemplate.opsForZSet().removeRangeByScore(key, start, end);
    }

    /**
     * 查询SET的size
     *
     * @param key
     */
    @Override
    public long zsetForSize(String key) {
        ZSetOperations setOperations = redisTemplate.opsForZSet();
        return setOperations.size(key);
    }

    /**
     * 获取ZSet全部元素
     *
     * @param key
     * @param reverse 是否需要相反排序获取元素
     * @param <T>
     * @return
     */
    @Override
    public <T> Set<T> zsetForAll(String key, Boolean reverse) {
        ZSetOperations setOperations = redisTemplate.opsForZSet();
        long length = setOperations.size(key);
        if (length <= 0) {
            return new HashSet();
        }
        if (reverse) {
            return setOperations.reverseRange(key, 0, -1);
        }
        return setOperations.range(key, 0, -1);
    }

    @Override
    public Long zsetRemoveRange(String key, Integer startIndex, Integer endIndex) {
        return redisTemplate.opsForZSet().removeRange(key, startIndex, endIndex);
    }

    @Override
    public <T> Set<T> zsetRangeByScore(String key, Double minScore, Double maxScore, Boolean isReverse) {
        if (isReverse) {
            return redisTemplate.opsForZSet().reverseRangeByScore(key, minScore, maxScore);
        }
        return redisTemplate.opsForZSet().rangeByScore(key, minScore, maxScore);
    }

    @Override
    public <T> Set<T> zsetRangeByScore(String key, Double minScore, Double maxScore, Long offset, Long count, Boolean isReverse) {
        if (isReverse) {
            return redisTemplate.opsForZSet().reverseRangeByScore(key, minScore, maxScore, offset, count);
        }
        return redisTemplate.opsForZSet().rangeByScore(key, minScore, maxScore, offset, count);
    }

    @Override
    public <T> List<LinkedHashSet<T>> zsetRangeByScorePipelined(List<String> keys, Double minScore, Double maxScore, Long offset, Long count, Boolean isReverse) {
        RedisSerializer keySerializer = redisTemplate.getKeySerializer();
        if (isReverse) {
            return redisTemplate.executePipelined((RedisCallback<T>) connection -> {
                keys.forEach(key -> connection.zRevRangeByScore(keySerializer.serialize(key), minScore, maxScore, offset, count));
                return null;
            });
        }
        return redisTemplate.executePipelined((RedisCallback<T>) connection -> {
            keys.forEach(key -> connection.zRangeByScore(keySerializer.serialize(key), minScore, maxScore, offset, count));
            return null;
        });
    }


    @Override
    public Set<ZSetOperations.TypedTuple<String>> getZsetCacheValuesScoresReverse(String key, Long start, Long end) {
        try {
            return redisTemplate.opsForZSet().reverseRangeWithScores(key, start, end);
        } catch (RuntimeException ex) {
            log.warn("get zset cache values with reverse scores fail. key:{}, start:{}, end:{}", key, start, end, ex);
        }
        return Collections.emptySet();
    }


    @Override
    public Set zsetReverseRangeByScore(String key, double min, double max) {
        return redisTemplate.opsForZSet().reverseRangeByScore(key, min, max);
    }

    @Override
    public boolean incrementByZsetCache(String key, String member, Long increasedScore) {
        return incrementByZsetCache(key, member, increasedScore.doubleValue());
    }

    @Override
    public boolean incrementByZsetCache(String key, String member, double increasedScore) {
        try {
            redisTemplate.opsForZSet().incrementScore(key, member, increasedScore);
            return true;
        } catch (RuntimeException ex) {
            log.warn("increment zset cache score fail. key:{}, increasedScore:{}", key, increasedScore, ex);
        }
        return false;
    }

    @Override
    public Long zsetForScoreCount(Object key, double min, double max) {
        return redisTemplate.opsForZSet().count(key, min, max);
    }

    @Override
    public Double getZsetCacheValueScore(String key, String value) {
        try {
            return redisTemplate.opsForZSet().score(key, value);
        } catch (RuntimeException ex) {
            log.warn("remvoe set value cache fail. key:{}, value:{}", key, value, ex);
        }
        return null;
    }

    /**
     * blocking 一直阻塞直到队列里边有数据
     * remove and get last item from queue:BRPOP
     *
     * @return
     */
    @Override
    public <T> T takeFromTail(String key, int timeout) throws InterruptedException {
//        lock.lockInterruptibly();
        RedisConnectionFactory connectionFactory = redisTemplate.getConnectionFactory();
        RedisConnection connection = connectionFactory.getConnection();
        try {
            byte[] rawKey = redisTemplate.getKeySerializer().serialize(key);
            List<byte[]> results = connection.bRPop(timeout, rawKey);
            if (CollectionUtils.isEmpty(results)) {
                return null;
            }
            return (T) redisTemplate.getValueSerializer().deserialize(results.get(1));
        } catch (Exception e) {
            log.error("获取队列信息异常:", e);
            return null;
        } finally {
//            lock.unlock();
            RedisConnectionUtils.releaseConnection(connection, connectionFactory);
        }
    }

    /**
     * blocking 一直阻塞直到队列里边有数据
     * remove and get first item from queue:BLPOP
     *
     * @return
     */
    @Override
    public <T> T takeFromHead(String key, int timeout) throws InterruptedException {
        //        lock.lockInterruptibly();
        RedisConnectionFactory connectionFactory = redisTemplate.getConnectionFactory();
        RedisConnection connection = connectionFactory.getConnection();
        try {
            byte[] rawKey = redisTemplate.getKeySerializer().serialize(key);
            List<byte[]> results = connection.bLPop(timeout, rawKey);
            if (CollectionUtils.isEmpty(results)) {
                return null;
            }
            return (T) redisTemplate.getValueSerializer().deserialize(results.get(1));
        } catch (Exception e) {
            log.error("获取队列信息异常:", e);
            return null;
        } finally {
//            lock.unlock();
            RedisConnectionUtils.releaseConnection(connection, connectionFactory);
        }
    }


    /**
     * 自增
     *
     * @param key
     * @return
     */
    @Override
    public long incr(String key) {
        return incr(key, 1);
    }


    /**
     * 递增
     *
     * @param key
     * @param delta 递增因子，必须大于0
     * @return
     */
    @Override
    public long incr(String key, long delta) {
        if (delta < 0) {
            throw new RuntimeException("递增因子必须大于0");
        }
        return redisTemplate.opsForValue().increment(key, delta);
    }

    /**
     * 自减
     *
     * @param key 键
     * @return
     */
    @Override
    public long decr(String key) {
        return decr(key, 1);
    }

    /**
     * 递减
     *
     * @param key   键
     * @param delta 要减少几(小于0)
     * @return
     */
    @Override
    public long decr(String key, long delta) {
        if (delta < 0) {
            throw new RuntimeException("递减因子必须大于0");
        }
        return redisTemplate.opsForValue().increment(key, -delta);
    }

    /**
     * 获取Integer类型的值
     *
     * @param key
     * @return
     */
    @Override
    public Integer getInteger(String key) {
        Object value = redisTemplate.opsForValue().get(key);
        return (value != null) ? (Integer) value : 0;
    }

    @Override
    public List hgetAllPipelined(List<String> keys) {
        RedisSerializer keySerializer = redisTemplate.getKeySerializer();
        return redisTemplate.executePipelined((RedisCallback<Object>) connection -> {
            keys.forEach(key -> connection.hGetAll(keySerializer.serialize(key)));
            return null;
        });
    }

    @Override
    public List getAllPipelined(List<String> keys) {
        RedisSerializer keySerializer = redisTemplate.getKeySerializer();
        return redisTemplate.executePipelined((RedisCallback<Object>) connection -> {
            keys.forEach(key -> connection.get(keySerializer.serialize(key)));
            return null;
        });
    }

    @Override
    public List smembersPipelined(List<String> keys) {
        RedisSerializer keySerializer = redisTemplate.getKeySerializer();
        return redisTemplate.executePipelined((RedisCallback<Object>) connection -> {
            keys.forEach(key -> {
                connection.sMembers(keySerializer.serialize(key));
            });
            return null;
        });
    }

    /**
     * 通配符查找redis的key
     *
     * @param parttern
     * @return
     */
    @Override
    public Set<String> keys(String parttern) {
        return redisTemplate.keys(parttern);
    }

    @Override
    public boolean tryLockMilliseconds(String key, long expireTime) {
        if (redisTemplate.opsForValue().setIfAbsent(key, LOCKED)) {
            redisTemplate.expire(key, expireTime, TimeUnit.MILLISECONDS);
            return true;
        }
        return false;
    }

    /**
     * 加锁
     *
     * @param key key值
     * @return 是否获取到
     */
    @Override
    public boolean disLock(String key, long expire) {
        String lock = LOCK_PREFIX + key;
        return (Boolean) redisTemplate.execute((RedisCallback) connection -> {
            long expireAt = System.currentTimeMillis() + expire + 1;
            Boolean acquire = connection.setNX(lock.getBytes(), String.valueOf(expireAt).getBytes());
            if (acquire) {
                return true;
            } else {
                byte[] value = connection.get(lock.getBytes());

                if (Objects.nonNull(value) && value.length > 0) {
                    long expireTime = Long.parseLong(new String(value));

                    if (expireTime < System.currentTimeMillis()) {
                        // 如果锁已经过期
                        byte[] oldValue = connection.getSet(lock.getBytes(), String.valueOf(System.currentTimeMillis() + expire + 1).getBytes());
                        // 防止死锁
                        return Long.parseLong(new String(oldValue)) < System.currentTimeMillis();
                    }
                }
            }
            return false;
        });
    }

    @Override
    public BoundListOperations boundListOps(String redisKey) {
        return redisTemplate.boundListOps(redisKey);
    }

    @Override
    public BoundHashOperations boundHashOps(String redisKey) {
        return redisTemplate.boundHashOps(redisKey);
    }

    @Override
    public List<String> mget(List<String> keys) {
        return redisTemplate.opsForValue().multiGet(keys);
    }

    @Override
    public Set<String> scan(String match) {

        StopWatch sw = new StopWatch("scan redis");
        sw.start("执行scan");
        Set<String> keySet = new HashSet<>();
        ScanOptions options = ScanOptions.scanOptions().match(match).count(DEFAULT_SCAN_COUNT).build();

        RedisSerializer<String> redisSerializer = (RedisSerializer<String>) redisTemplate.getKeySerializer();
        Cursor<String> cursor = (Cursor) redisTemplate.executeWithStickyConnection(redisConnection ->
                new ConvertingCursor<>(redisConnection.scan(options), redisSerializer::deserialize));
        sw.stop();

        sw.start("封装参数");
        while (cursor.hasNext()) {
            keySet.add(cursor.next());
        }
        sw.stop();

        sw.start("关闭指针");
        try {
            cursor.close();
        } catch (Exception e) {
            log.error("关闭 redis cursor 失败");
        }
        sw.stop();
        log.info(sw.prettyPrint());
        return keySet;
    }
}
